package com.lxz.rpcbasednetty;

import java.util.HashMap;
import java.util.Map;

import org.apache.commons.logging.LogFactory;
import org.jboss.netty.handler.codec.serialization.ClassResolvers;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cglib.reflect.FastClass;
import org.springframework.cglib.reflect.FastMethod;

import com.distribute.rpc.sample.server.HelloServiceImpl;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.serialization.ClassResolver;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;

public class RpcServer {
	private static final Logger LOGGER = LoggerFactory.getLogger(RpcServer.class);
	private final int port;
	
	public RpcServer(int port){
		this.port = port;
	}
	
	public void startServer() throws Exception{
		//配置服务端的NIO线程组
		EventLoopGroup bossGroup = new NioEventLoopGroup();
		EventLoopGroup workerGroup = new NioEventLoopGroup();
		try{
			ServerBootstrap b = new ServerBootstrap();
			b.group(bossGroup, workerGroup)
				.channel(NioServerSocketChannel.class)
				.option(ChannelOption.SO_BACKLOG, 1024)
				.childHandler(new ChildChannelHandler());
			//绑定端口，同步等待成功
			ChannelFuture f = b.bind(port).sync();
			//等待服务端监听端口关闭
			f.channel().closeFuture().sync();
		}finally{
			bossGroup.shutdownGracefully();
			workerGroup.shutdownGracefully();
		}
	}
		
	private class ChildChannelHandler extends ChannelInitializer<SocketChannel>{
		@Override
		protected void initChannel(SocketChannel ch) throws Exception {
			ch.pipeline().addLast(new ObjectDecoder((ClassResolver) ClassResolvers.cacheDisabled(this.getClass().getClassLoader())))
					.addLast(new ObjectEncoder())
					.addLast(new RpcServerHandler());
		}
	}
	
	private class RpcServerHandler extends ChannelHandlerAdapter{
		private final Map<String, Object> handlerMap = new HashMap<String, Object>();
		
		public RpcServerHandler(){
			handlerMap.put("HelloService", HelloServiceImpl.class);
		}
		
		public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception{	
			RpcRequest request = (RpcRequest) msg;
			RpcResponse response = new RpcResponse();
			response.setRequestId(request.getRequestId());
			try{
				Object result = handle(request);
				response.setResult(result);
			}catch(Throwable t){
				response.setError(t);
			}
			ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
		}
		
		private Object handle(RpcRequest request) throws Throwable{
			String className = request.getClassName();
			Object serviceBean = handlerMap.get(className);
			
			Class<?> serviceClass = serviceBean.getClass();
			String methodName = request.getMethodName();
			Class<?>[] parameterTypes = request.getParameterTypes();
			Object[] parameters = request.getParameters();
			
			FastClass serviceFastClass = FastClass.create(serviceClass);
			FastMethod serviceFastMethod = serviceFastClass.getMethod(methodName, parameterTypes);
			return serviceFastMethod.invoke(serviceBean, parameters);
		}
		
		public void channelReadComplete(ChannelHandlerContext ctx){
			ctx.flush();
		}
		
		public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause){
			LOGGER.error("Unexcepted error", cause);
			ctx.close();
		}
	}
}
